#ifndef HEARTBEAT_HPP
#define HEARTBEAT_HPP
#include "cache_server.hpp"
class taskconnect_class{
private:
    LRUCache *LC;
    std::string addr;
    std::string ip_oth;
    int port_oth;
    struct sockaddr_in cache_oth_addr;
    std::unordered_map<std::string, std::vector<std::string>> *ipport_and_key;
    int cachfd;
    int get_ipport(std::string& ip, std::string& addr);
public:
    taskconnect_class(task_connect_oth *task_arg);
    ~taskconnect_class();
    void net_connect();
    void sendout_data();
};
class iplistchange_class{
private:
    ConsistentHash *key_addr;
    std::vector<std::string> *ipport_list;
    ThreadPool *ThrPl;
    LRUCache *LC;
    int cfd;
    std::unordered_map<std::string, std::vector<std::string>> ipport_and_key;
    struct task_connect_oth task_arg;
    std::string curen_addr = CACHESERV_IP+(std::string)":"+std::to_string(CACHESERV_PORT);
    bool exit_flag;
public:
    iplistchange_class(struct iplistchange_struct *iplistchage);
    ~iplistchange_class();
    void organize_data();
    void takeout_alldata();
};
class Heartbeat{
private:    
    ConsistentHash *key_addr;
    std::vector<std::string> *ipport_list;
    ThreadPool *ThrPl;
    LRUCache *LC;
    struct sockaddr_in serv_addr;
    json heartbeat_json, iplist_json;
    char buf_rdlist[BUFSIZ];
    std::string buf;
    int cfd;
    struct iplistchange_struct ipchange_arg;
public:
    Heartbeat(struct heartbeat_struct *heartbeat_arg);
    ~Heartbeat();
    void heartbeat_connect();
    void heartbeat_send();
    int is_anycommond();
    void shutdown_handler();
    void add_handler();
    void reflesh_handler();
};

void *task_connectoth(void *arg){
    pthread_mutex_lock(&transfertooth);
    taskconnect_class taskconnect((struct task_connect_oth *)arg);
    pthread_mutex_unlock(&transfertooth);
    taskconnect.net_connect();
    taskconnect.sendout_data();
    return nullptr;
}
void* doit_iplistchange(void *arg){
    pthread_rwlock_wrlock(&ipchange_lock);
    iplistchange_class ipchange((struct iplistchange_struct *)arg);
//take out all my data to other caches
    ipchange.organize_data();
    ipchange.takeout_alldata();
    pthread_rwlock_unlock(&ipchange_lock);
    return nullptr;
}
json Write_oth_ClientCache(std::string& key, std::string& value){
    json client_to_cache, data;
    data["flag"] = true;
    data["key"] = key;
    data["value"] = value;
    client_to_cache["type"] = KEY_VALUE_RESPOND;
    client_to_cache["data"] = data;
    return client_to_cache;
}
json pack_json_heartbeat(){
    json data, heartbeat_json;
    std::string number = CACHESERV_IP+(const std::string)":"+std::to_string(CACHESERV_PORT);
    data["iplist"] = number;
    data["state"] = true;
    heartbeat_json["type"] = HEART_BEAT;
    heartbeat_json["data"] = data;
    return heartbeat_json;
}


taskconnect_class::taskconnect_class(task_connect_oth *task_arg){
    LC = task_arg->LC;
    addr = task_arg->addr;
    ip_oth.clear();
    ipport_and_key = task_arg->ipport_and_key;
}
taskconnect_class::~taskconnect_class(){

}
int taskconnect_class::get_ipport(std::string& ip, std::string& addr){
    bool ip_order = true;
    std::string port("");
    for(char c:addr){
        if(c==':'){
            ip_order = !ip_order;
            continue;
        }
        if(ip_order)
            ip += c;
        else
            port+=c;
    }
    return std::stoi(port);
}
void taskconnect_class::net_connect(){
    port_oth = get_ipport(ip_oth, addr);
    cachfd = Socket(AF_INET, SOCK_STREAM, 0);
    bzero(&cache_oth_addr,sizeof(cache_oth_addr));
    cache_oth_addr.sin_family = AF_INET;
    inet_pton(AF_INET, (const char*)ip_oth.data(), &cache_oth_addr.sin_addr.s_addr);
    cache_oth_addr.sin_port = htons(port_oth);
    Connect(cachfd, (struct sockaddr *)&cache_oth_addr, sizeof(cache_oth_addr));
}
void taskconnect_class::sendout_data(){
    for(int j=0; j<(*ipport_and_key)[addr].size(); j++){
    std::string key = (*ipport_and_key)[addr][j];
    std::string value = (*LC).get(key);
    json write_j = Write_oth_ClientCache(key, value);
    std::string buf = write_j.dump();
    Write(cachfd, (const char *)buf.data(), BUFSIZ);
    Sleep_ms(200);
    }
    Close(cachfd);
}

iplistchange_class::iplistchange_class(struct iplistchange_struct *iplistchage){
    key_addr = iplistchage->key_addr;
    ipport_list = iplistchage->ipport_list;
    ThrPl = iplistchage->ThrPl;
    LC = iplistchage->LC;
    cfd = iplistchage->cfd;
    (*key_addr).RefreshIPList(*ipport_list);
    exit_flag = iplistchage->exit_flag;
    if(exit_flag)
        std::cout<<"prepare to close..."<<std::endl;  
}
iplistchange_class::~iplistchange_class(){
}
void iplistchange_class::organize_data(){
    for(std::unordered_map<std::string, DLinkedNode*>::iterator i = \
        (*LC).cache.begin(); i!=(*LC).cache.end();i++){
    std::string key = i->first;
    std::string addr = (*key_addr).GetServerIndex(key);
    std::cout<<addr<<std::endl;
    ipport_and_key[addr].push_back(key);
    }
}
void iplistchange_class::takeout_alldata(){
    signal(SIGPIPE, SIG_IGN);
    task_arg.LC = LC;
    task_arg.ipport_and_key = &ipport_and_key;
    std::cout<<"============="<<std::endl;
    pthread_t takeout_allval[ipport_and_key.size()];
    int count=0;
    for(std::unordered_map<std::string, std::vector<std::string>>::iterator i = ipport_and_key.begin(); \
        i!=ipport_and_key.end(); i++){
        Sleep_ms(200);
        if(curen_addr==i->first)
            continue;
        pthread_mutex_lock(&transfertooth);
        task_arg.addr = i->first;
        std::cout<<i->first<<std::endl;//test!!!!!!!!!!!!!!!!!
        Pthread_create(&takeout_allval[count++], nullptr, &task_connectoth, &task_arg);
        pthread_mutex_unlock(&transfertooth);        
    }
    Sleep_ms(200);
    for(int i=0; i<ipport_and_key.size(); i++)
        pthread_join(takeout_allval[i], nullptr);
    if(exit_flag){
        std::cout<<"close done"<<std::endl;
        exit(0);
    }
}
Heartbeat::Heartbeat(struct heartbeat_struct *heartbeat_arg){
    key_addr = heartbeat_arg->key_addr;
    ipport_list = heartbeat_arg->ipport_list;
    ThrPl = heartbeat_arg->ThrPl;
    LC = heartbeat_arg->LC;
    heartbeat_json = pack_json_heartbeat();
}
Heartbeat::~Heartbeat(){
    pthread_rwlock_unlock(&ipchange_lock);
    pthread_rwlock_destroy(&ipchange_lock);
    pthread_mutex_destroy(&transfertooth);
    Close(cfd);
}
void Heartbeat::heartbeat_connect(){
    signal(SIGPIPE, SIG_IGN);
    bzero(&serv_addr,sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    inet_pton(AF_INET, MASTER_IP, &serv_addr.sin_addr.s_addr);
    serv_addr.sin_port = htons(MASTER_PORT);
    buf = heartbeat_json.dump();
    cfd = Socket_connect(true, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
}
void Heartbeat::heartbeat_send(){
    int n;
    buf += "\0";
    ////important!!!!!!!!!!!!!!!!!!!!!!!!
    n = Write(cfd, (char *)buf.data(), buf.length()+1);
    if(n<0){
       cfd = Socket_connect(false, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
    }
}
int Heartbeat::is_anycommond(){
    int m=0;
    do{
        m = unblock_read_net(cfd, buf_rdlist, BUFSIZ);
        if(0<m){
            iplist_json = json::parse(buf_rdlist);
            std::cout<<buf_rdlist<<std::endl;//test!!!!!!!!!!!!!!!!!!!!!!!
            return (int)iplist_json["type"];
        }
        else if(0==m)
            cfd = Socket_connect(false, (struct sockaddr *)&serv_addr, sizeof(serv_addr));
    }while(0==m);
    return -1;
}
void Heartbeat::shutdown_handler(){
    (*ipport_list).clear();
    for(int i=0;i<iplist_json["data"]["iplist"].size();i++){
        (*ipport_list).push_back(iplist_json["data"]["iplist"][i]);
    }
    pthread_t tid_beforeshutdown;
    ipchange_arg.ThrPl = ThrPl;
    ipchange_arg.LC = LC;
    ipchange_arg.exit_flag = true;
    ipchange_arg.key_addr = key_addr;
    ipchange_arg.ipport_list = ipport_list;
    Pthread_create(&tid_beforeshutdown, nullptr, doit_iplistchange, &ipchange_arg);
}
void Heartbeat::add_handler(){
    (*ipport_list).clear();
    for(int i=0;i<iplist_json["data"]["iplist"].size();i++){
        (*ipport_list).push_back(iplist_json["data"]["iplist"][i]);
    }
    pthread_t tid_askfromothercache;
    ipchange_arg.ThrPl = ThrPl;
    ipchange_arg.LC = LC;
    ipchange_arg.exit_flag = false;
    ipchange_arg.key_addr = key_addr;
    ipchange_arg.ipport_list = ipport_list;
    Pthread_create(&tid_askfromothercache, nullptr, doit_iplistchange, &ipchange_arg);
}
void Heartbeat::reflesh_handler(){
    (*ipport_list).clear();
    for(int i=0;i<iplist_json["data"]["iplist"].size();i++){
        (*ipport_list).push_back(iplist_json["data"]["iplist"][i]);
    }
    (*key_addr).RefreshIPList(*ipport_list);
}


#endif /* HEARTBEAT_HPP */
